#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <sys/mman.h>
#include <assert.h>
#include <pthread.h>
#include <fcntl.h>
#include <stdbool.h>
#include "MobsBuf.h"

#define DEBUG 1

#if DEBUG
#define LOG(fmt,args...) printf("[mods %s:%d]"fmt,__FUNCTION__,__LINE__,##args)
#else
#define LOG(fmt,args...)
#endif

#define MMAP_MIN_PAGE_SIZE (4096)
#define __ALIGN_MASK(x,mask)    (((x)+(mask))&~(mask))
#define ALIGN(x,a)              __ALIGN_MASK(x,(__typeof__(x))(a)-1) 
#define MAX_CLASS_USER_NUM  (16)
#define MAX_USER_NAME_SIZE  (32)

typedef enum object_user_role_e
{
    ROLE_UNKNOWND = 0x00,
    ROLE_PRODUCER = 0x01,
    ROLE_CONSUMER ,
    ROLE_MAINTAINER ,
    ROLE_WATCHER ,
}ObjectUserRole;
typedef struct object_user_attr_t
{
    uint8_t role;
    uint8_t state;
    uint16_t objIdx; 
    uint32_t datOffset; // used in read op
    void* mmapaddr;
    char name[MAX_USER_NAME_SIZE];// we need name to solve user's dead and restarted caused problem!
}ObjectUserAttr;

#define MOBS_MAGIC "MOBS"
typedef struct class_block_attr_t
{
    uint32_t magic;// "MOBS"
    uint32_t flags; // open mode,only saved buffer global flags
    uint16_t object_max_num; // determine object-region-size,dynamic allocate when opened
    uint16_t object_first_readIdx;
    uint16_t object_free_startIdx; // to write at this node
    uint32_t object_zone_offset; // = sizeof(ClassBlockAttr) + data table length
    uint32_t data_zone_offset;// = object_zone_offset + sizeof(ObjectBlockAttr)*object_max_num

    pthread_mutex_t lock;

    uint32_t object_expire_strategy;// 1B Method,3B Value
    uint32_t object_user_num;
    char class_fullpath[64];
    ObjectUserAttr users[MAX_CLASS_USER_NUM];

    uint32_t data_page_size;// 1KB,2KB,4KB(default),8KB
    uint32_t data_page_total_num;// depend on user's class-buf size
    uint32_t data_page_free_num;// determine available data buffer size = free_num * page_size
    uint32_t data_page_free_startIdx; // to lookup allocate_table[] get free-page-list
    uint32_t data_page_allocate_table[0]; // depend on @data_page_total_num@
}ClassBlockAttr;

#define OBJECT_FLAG_UNUSED  0x00
#define OBJECT_FLAG_COVERED 0x01
#define OBJECT_FLAG_INCOMPLETE 0x02
#define OBJECT_FLAG_COMPLETED 0x04
typedef struct object_block_attr_t
{
    uint8_t state;
    uint8_t refCnt;
    uint32_t data_size;
    uint32_t data_startIdx;
    uint32_t data_endIdx;
    uint32_t data_idxNum;
    uint32_t created_at;
}ObjectBlockAttr;

#define CONVERT_UID_TO_PTR(uid) \
    if(uid==0) return MOBS_ERR_PARAM; \
	ObjectUserAttr *puser = (ObjectUserAttr*)uid; \
	if(puser->role == ROLE_UNKNOWND || puser->state!=1 || puser->mmapaddr==NULL) \
	{ return -1;} \
	uint8_t *pHeadAddr = (uint8_t*)puser->mmapaddr; \
	ClassBlockAttr* pclass = (ClassBlockAttr*)puser->mmapaddr;

#define CHECK_OBJECT_ID(oid) \
    if(oid < 0 || oid >= pclass->object_max_num) return MOBS_ERR_PARAM;

// for multi-thread in one process to open same bufClass
static pthread_mutex_t s_buf_mutex = PTHREAD_MUTEX_INITIALIZER;
// Lookup Buffer Table for multi-thread in one process: MOBS_FLAG_THREADS
#if 0
#define ONEPROCESS_MAX_CLASS_NUM 64
static struct lookup_buffer_table
{// only used in MOBS_FLAG_THREADS for MAP_PRIVATE
    char className[64];
    uintptr_t classEntry;
}LookupTable[ONEPROCESS_MAX_CLASS_NUM];
#else
typedef struct buffer_class_item
{
    char name[64];
    uintptr_t entry;
    struct buffer_class_item *pnext;
}BufferClassItem;
static BufferClassItem *LookupTable = NULL;
#endif

static uintptr_t find_class_item(const char *buf_name)
{
    //pthread_mutex_lock(&s_buf_mutex);
    BufferClassItem *p = LookupTable;
    uintptr_t pItem = 0;
    while(p!=NULL)
    {
	if(strcmp(buf_name,p->name)==0)
	{
	    pItem = p->entry;
	    break;
	}
	p = p->pnext;
    }
    //pthread_mutex_unlock(&s_buf_mutex);
    return pItem;
}
static bool add_class_item(const char *buf_name,uintptr_t entry)
{
    bool ok = false,find = false;
    //pthread_mutex_lock(&s_buf_mutex);
    BufferClassItem *p = LookupTable;
    while(p!=NULL)
    {
	if(strcmp(buf_name,p->name)==0)
	{
	    find = true;
	    break;
	}
	p = p->pnext;
    }
    if(!find)
    {
	BufferClassItem *pnew = (BufferClassItem*)malloc(sizeof(BufferClassItem));
	if(pnew!=NULL)
	{
	    strncpy(pnew->name,buf_name,sizeof(pnew->name)-1);
	    pnew->entry = entry;
	    pnew->pnext = LookupTable;
	    LookupTable = pnew;
	    ok = true;
	}
    }
    //pthread_mutex_unlock(&s_buf_mutex);
    return ok;
}
static bool del_class_item(const char *buf_name)
{
    bool ok = false;
    //pthread_mutex_lock(&s_buf_mutex);
    BufferClassItem *p = LookupTable;
    BufferClassItem *prev = NULL;
    while(p!=NULL)
    {
	if(strcmp(buf_name,p->name)==0)
	{
	    ok = true;
	    if(prev != NULL)
	    {
		prev->pnext = p->pnext;
	    }
	    else
	    {
		LookupTable = p->pnext;
	    }
	    free(p);p = NULL;
	    break;
	}
	p = p->pnext;
	prev = p;
    }
    //pthread_mutex_unlock(&s_buf_mutex);
    return ok;
}

static void _lock(ClassBlockAttr *pclass)
{
    int ret = pthread_mutex_lock(&pclass->lock);
    if(ret == EOWNERDEAD)
    {
	LOG("lock owner is dead!\n");
	pthread_mutex_consistent(&pclass->lock);
    }
}
static void _unlock(ClassBlockAttr*pclass)
{
    pthread_mutex_unlock(&pclass->lock);
}

uintptr_t mobs_open(const char *user_name,const char *buf_name,int flags,size_t size,size_t max_object_num,size_t data_page_size)
{
    if(NULL==user_name || user_name[0]==0 || strlen(user_name)>=MAX_USER_NAME_SIZE)
    {
	LOG("invalid user_name!\n");
	return 0;
    }
    if(NULL==buf_name || buf_name[0]!='/')
    {
	LOG("invalid buf_name:%p\n",buf_name);
	return 0;
    }
    if(size==0 || size <= data_page_size)
    {
	LOG("invalid buf size or pagesize:%lu,%lu\n",size,data_page_size);
	return 0;
    }
    int _rdwrFlag = flags & (MOBS_FLAG_RDONLY|MOBS_FLAG_WRONLY);
    if(_rdwrFlag == (MOBS_FLAG_RDONLY|MOBS_FLAG_WRONLY) || _rdwrFlag == 0)
    {// both or none err
	LOG("flags:0x%08x err!\n",flags);
	return 0;
    }
    int _modeFlag = flags & (MOBS_FLAG_SHARED|MOBS_FLAG_PRIVATE);
    if(_modeFlag == (MOBS_FLAG_SHARED|MOBS_FLAG_PRIVATE) || _modeFlag == 0)
    {// both or none err
	LOG("flags:0x%08x err,shared or private only one and must one!\n",flags);
	return 0;
    }
    int _consumeFlag = flags & (MOBS_FLAG_LATEST|MOBS_FLAG_OLDEST);
    if(_consumeFlag == (MOBS_FLAG_LATEST|MOBS_FLAG_OLDEST))
    {
	LOG("flags:0x%08x err,latest or oldest only one!\n",flags);
	return 0;
    }
    if(_modeFlag == MOBS_FLAG_PRIVATE)
    {// in Private workmode consumer must deal obejct from oldest to maintain class first_readIdx proceed!
	_consumeFlag = MOBS_FLAG_OLDEST;
    }
    if(flags&MOBS_FLAG_THREADS)
    {
	_modeFlag |= MOBS_FLAG_THREADS;
    }
    else
    {
	_modeFlag |= MOBS_FLAG_PROCESS;
    }

    if(max_object_num==0) max_object_num = 4096;
    if(data_page_size==0) data_page_size = 4096;

    data_page_size = ALIGN(data_page_size,1024);
    uint32_t data_buf_size = ALIGN(size,MMAP_MIN_PAGE_SIZE);
    uint32_t data_page_num = data_buf_size / data_page_size; 
    uint32_t data_buf_offset = sizeof(ClassBlockAttr) + sizeof(uint32_t)*data_page_num + sizeof(ObjectBlockAttr)*max_object_num;
    data_buf_offset = ALIGN(data_buf_offset,MMAP_MIN_PAGE_SIZE);

    uint32_t total_buf_size = data_buf_offset + data_buf_size;
    uint8_t *address = NULL;
    if(flags & MOBS_FLAG_THREADS)
    {
	address = (uint8_t*)find_class_item(buf_name);
    }
    if(address==NULL)
    {
	address = (uint8_t*)mmap(NULL,total_buf_size,PROT_READ|PROT_WRITE,MAP_ANONYMOUS|MAP_PRIVATE,-1,0);
	if(address == MAP_FAILED)
	{
	    LOG("mmap size %u failed!\n",total_buf_size);
	    return 0;
	}
    }
    LOG("data page size:%lu,buf size: %u,page num:%u,buf offset:%u,total buf size:%u,address:%p\n",
	    data_page_size,data_buf_size,data_page_num,data_buf_offset,total_buf_size,address);

    ObjectUserAttr *puser = NULL;// for goto
    pthread_mutex_lock(&s_buf_mutex);
    int fd = open(buf_name,O_CREAT|O_RDWR,S_IRUSR|S_IWUSR);
    if(fd < 0)
    {
	LOG("open file %s fail:%s\n",buf_name,strerror(errno));
	pthread_mutex_unlock(&s_buf_mutex);
	return 0;
    }
    if(-1 == ftruncate(fd,total_buf_size))
    {
	LOG("truncate file %s to %u fail:%s\n",buf_name,total_buf_size,strerror(errno));
	goto open_end;
    }
    struct flock buf_lock;
    buf_lock.l_type = F_WRLCK;
    buf_lock.l_whence = SEEK_SET;
    buf_lock.l_start = 0;
    buf_lock.l_len = total_buf_size;
    if(fcntl(fd,F_SETLKW,&buf_lock)==-1)
    {
	LOG("fcntl lock buf file fail!\n");
	goto open_end;
    }

    /*
     * Note: jffs2 filesystem don't support MAP_SHARED ,only MAP_PRIVATE is valid option!!
     *
     */
    size_t i;
    int prot = PROT_READ;
    int mapFlags = MAP_SHARED|MAP_FIXED;
    uint8_t *pHeadAddr = address;
    if(flags & MOBS_FLAG_THREADS)
    {
	mapFlags = MAP_PRIVATE|MAP_FIXED;
    }
    else
    {
	pHeadAddr = (uint8_t*)mmap(address,data_buf_offset, PROT_READ|PROT_WRITE,mapFlags,fd,0);
	if(MAP_FAILED==pHeadAddr || pHeadAddr != address)
	{
	    LOG("mmap head fail address %p,size %u,%s\n",address,data_buf_offset,strerror(errno));
	    goto open_end;
	}

	if(flags & MOBS_FLAG_WRONLY)
	{
	    prot = PROT_WRITE;
	}
	uint8_t *pDataAddr = (uint8_t*)mmap(address+data_buf_offset,data_buf_size, prot,mapFlags,fd,data_buf_offset);
	if(MAP_FAILED==pDataAddr || pDataAddr!=address+data_buf_offset)
	{
	    LOG("mmap data fail address %p,size %u,%s\n",
		    address+data_buf_offset,data_buf_size,strerror(errno));
	    goto open_end;
	}

	LOG("mmap fd %d head addr:%p,data addr:%p\n",fd,pHeadAddr,pDataAddr);
    }

    ClassBlockAttr *pClassAttr = (ClassBlockAttr*)pHeadAddr;
    if(memcmp(&pClassAttr->magic,MOBS_MAGIC,sizeof(pClassAttr->magic))==0)
    {
	if(pClassAttr->data_page_total_num != data_page_num || 
		pClassAttr->data_zone_offset != data_buf_offset ||
		pClassAttr->data_page_size != data_page_size ||
		pClassAttr->object_max_num != max_object_num)
	{
	    LOG("config incorrect,has data page num:%u,zone offset:%u,max objnum %u,data page size %u\n",
		    pClassAttr->data_page_total_num,pClassAttr->data_zone_offset,pClassAttr->object_max_num,pClassAttr->data_page_size);
	    LOG("   now set data page num:%u,zone offset:%u,max objnum %lu,data page size %lu\n",
		    data_page_num,data_buf_offset,max_object_num,data_page_size);
	    goto open_end;
	}
	if(strcmp(pClassAttr->class_fullpath,buf_name)!=0)
	{
	    LOG("config path incorrect,has :%s,now open:%s\n",pClassAttr->class_fullpath,buf_name);
	    goto open_end;
	}
	if(pClassAttr->flags != (_modeFlag))
	{
	    LOG("config workmode incorrect,has :0x%08x,now open:0x%08x\n",pClassAttr->flags,_modeFlag);
	    goto open_end;
	}
    }
    else
    {
	LOG("Init New Class data page num:%u,zone offset:%u,max objnum %lu,data page size %lu\n",
		data_page_num,data_buf_offset,max_object_num,data_page_size);
	// init mutex lock
	pthread_mutexattr_t mattr;
	pthread_mutexattr_init(&mattr);
	pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
	pthread_mutexattr_settype(&mattr, PTHREAD_MUTEX_RECURSIVE);
	pthread_mutexattr_setrobust(&mattr, PTHREAD_MUTEX_ROBUST);
	pthread_mutex_init(&pClassAttr->lock,&mattr);

	pClassAttr->object_max_num = max_object_num;
	pClassAttr->object_user_num = 0;
	pClassAttr->object_free_startIdx = 0;
	pClassAttr->object_expire_strategy = 0x00001000;
	pClassAttr->data_page_size = data_page_size;
	pClassAttr->data_page_total_num = data_page_num;
	pClassAttr->data_page_free_num = data_page_num;
	pClassAttr->data_page_free_startIdx = 0;
	pClassAttr->object_zone_offset = sizeof(ClassBlockAttr) + sizeof(uint32_t)*data_page_num;
	pClassAttr->data_zone_offset = data_buf_offset;

	snprintf(pClassAttr->class_fullpath,sizeof(pClassAttr->class_fullpath),"%s",buf_name);

	// init data page table
	uint32_t *pTable = pClassAttr->data_page_allocate_table;
	for(i=0;i<data_page_num;i++)
	{
	    pTable[i] = (i+1);// all free page, data_page_num indicates end!
	}

	// init object item block
	ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)(pHeadAddr+pClassAttr->object_zone_offset);
	memset(pObjBlk,0,sizeof(ObjectBlockAttr)*max_object_num);

	pClassAttr->flags = _modeFlag;
	memset(pClassAttr->users,0,sizeof(pClassAttr->users));
	memcpy(&pClassAttr->magic,MOBS_MAGIC,sizeof(pClassAttr->magic));

	if(flags & MOBS_FLAG_THREADS)
	{
	    add_class_item(buf_name,(uintptr_t)pHeadAddr);
	}
    }

    int32_t slot = -1;
    int32_t first_empty_slot = -1;
    int32_t existed_user_slot = -1;
    for(i=0;i<MAX_CLASS_USER_NUM;i++)
    {
	if(first_empty_slot < 0 && pClassAttr->users[i].role == ROLE_UNKNOWND)
	{
	    first_empty_slot = i;
	}
	if(existed_user_slot < 0 && strcmp(pClassAttr->users[i].name,user_name)==0)
	{
	    existed_user_slot = i;
	}
    }
    slot = (existed_user_slot >= 0)?existed_user_slot:first_empty_slot;
    if(slot >= 0)
    {
	i = slot;
	puser = &pClassAttr->users[i];
	if(flags & MOBS_FLAG_WRONLY)
	{
	    LOG("producer user slot %u,now user num %u\n",slot,pClassAttr->object_user_num);
	    puser->role = ROLE_PRODUCER;
	    puser->state = 1;
	    puser->objIdx = pClassAttr->object_free_startIdx;
	    puser->datOffset = 0;
	    puser->mmapaddr = pHeadAddr;
	    snprintf(puser->name,sizeof(puser->name),"%s",user_name);
	    if(slot != existed_user_slot) pClassAttr->object_user_num ++;

	    goto open_end;
	}
	else
	{
	    LOG("consumer user slot %u,now user num %u\n",slot,pClassAttr->object_user_num);
	    puser->role = ROLE_CONSUMER;
	    puser->state = 1;

	    if((flags & MOBS_FLAG_CONTINUE) && (slot == existed_user_slot))
	    {
		// keep last objIdx for breakpoint resume
	    }
	    else if(_consumeFlag == MOBS_FLAG_OLDEST)
	    {
		puser->objIdx = pClassAttr->object_first_readIdx;
	    }
	    else
	    {
		puser->objIdx = pClassAttr->object_free_startIdx;
	    }

	    puser->datOffset = 0;
	    puser->mmapaddr = pHeadAddr;
	    snprintf(puser->name,sizeof(puser->name),"%s",user_name);
	    if(slot != existed_user_slot) pClassAttr->object_user_num ++;

	    goto open_end;
	}
    }
open_end:
    close(fd);
    pthread_mutex_unlock(&s_buf_mutex);
    return (uintptr_t)puser;
}

int mobs_close(uintptr_t uid)
{
    CONVERT_UID_TO_PTR(uid);

    _lock(pclass);
    uint8_t *address = (uint8_t*)puser->mmapaddr;
    memset(puser,0,sizeof(ObjectUserAttr));
    pclass->object_user_num-- ;

    // last user unlink class file
    if(pclass->object_user_num == 0)
    {
	unlink(pclass->class_fullpath);
    }
    _unlock(pclass); 

    pthread_mutex_lock(&s_buf_mutex);
    del_class_item(pclass->class_fullpath);
    pthread_mutex_unlock(&s_buf_mutex);

    if(pclass->flags & MOBS_FLAG_PROCESS)
    {
	munmap(address,pclass->data_zone_offset+pclass->data_page_size*pclass->data_page_total_num);
    }

    return 0;
}

int mobs_reset(uintptr_t uid)
{
    CONVERT_UID_TO_PTR(uid);

    _lock(pclass);
    puser->objIdx = pclass->object_free_startIdx;
    puser->datOffset = 0;
    _unlock(pclass);
    return 0;
}

int mobs_flush(uintptr_t uid)
{
    CONVERT_UID_TO_PTR(uid);

    int ret = msync((void*)puser->mmapaddr,
	pclass->data_zone_offset+pclass->data_page_size*pclass->data_page_total_num,
	MS_ASYNC);
    return ret;
}

int mobs_unlink(const char *buf_name)
{
    if(buf_name==NULL || strlen(buf_name)<=0)
	return -1;

    char cmd[64]={0};
    snprintf(cmd,sizeof(cmd),"lsof -t %s > /dev/null 2>&1",buf_name);// find if has other process open the file

    int ret = system(cmd);
    if(ret == -1) return -1;

    int code = WEXITSTATUS(ret);
    if(ret != 0)// 0 - has some process is using it
    {// if there's no process use it,we are creater so unlink it to avoid use error-residual-class-file!
	unlink(buf_name);
    }
    return 0;
}

int get_available_object_num(ClassBlockAttr *pclass,int toIdx)
{
    if(toIdx < 0) toIdx = pclass->object_first_readIdx;

    if(toIdx == pclass->object_free_startIdx)
    {
	// Empty object available
	return 0;
    }
    uint32_t available_num = 0;
    if(pclass->object_free_startIdx < toIdx)
    {
	available_num = pclass->object_free_startIdx + pclass->object_max_num - toIdx;
    }
    else
    {
	available_num = pclass->object_free_startIdx - toIdx;
    }
    return available_num;
}
static void delete_object(ClassBlockAttr* pclass,int objIdx)
{
    if(objIdx < 0) objIdx = pclass->object_first_readIdx;

    ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)((uint8_t*)pclass+ pclass->object_zone_offset);
    ObjectBlockAttr *pObj = &pObjBlk[objIdx];

    if(pObj->state != OBJECT_FLAG_COMPLETED)
    {
	LOG("object %d state not competed 0x%x\n",objIdx,pObj->state);
	return;
    }

    pObj->state = OBJECT_FLAG_COVERED;
    pObj->refCnt = 0;
    if(pObj->data_size > 0)
    {// release data page to table,add this list to free list-head
	pclass->data_page_allocate_table[pObj->data_endIdx] = pclass->data_page_free_startIdx;
	pclass->data_page_free_startIdx = pObj->data_startIdx;
	pclass->data_page_free_num += pObj->data_idxNum;
    }
    pObj->data_size = 0;
    pObj->data_startIdx = 0;
    pObj->data_endIdx = 0;
    pObj->data_idxNum = 0;

    if(objIdx == pclass->object_first_readIdx)
    {
	pclass->object_first_readIdx = (pclass->object_first_readIdx+1) % pclass->object_max_num;
    }
}

int mobs_start(uintptr_t uid,int flags)
{
    CONVERT_UID_TO_PTR(uid);

    int oid = -1;

    _lock(pclass);
    ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)(pHeadAddr + pclass->object_zone_offset);
    if(puser->role == ROLE_PRODUCER)
    {
	puser->objIdx = pclass->object_free_startIdx;
	puser->datOffset = 0;
	pclass->object_free_startIdx = (pclass->object_free_startIdx+1) % pclass->object_max_num;
	if(pclass->object_free_startIdx == pclass->object_first_readIdx || pclass->data_page_free_num == 0)
	{// full object list,delete one
	    delete_object(pclass,-1);
	}
	LOG("puser %p objIdx %hu,class next free Idx %hu\n", puser,puser->objIdx,pclass->object_free_startIdx);

	pObjBlk[puser->objIdx].state = OBJECT_FLAG_INCOMPLETE;
	pObjBlk[puser->objIdx].refCnt++;
	pObjBlk[puser->objIdx].data_size = 0;
	pObjBlk[puser->objIdx].data_startIdx = pclass->data_page_free_startIdx;
	pObjBlk[puser->objIdx].data_endIdx = pObjBlk[puser->objIdx].data_startIdx;
	pObjBlk[puser->objIdx].data_idxNum = 0;
	pObjBlk[puser->objIdx].created_at = time(NULL);
	pclass->data_page_free_startIdx = pclass->data_page_allocate_table[pclass->data_page_free_startIdx];
	pclass->data_page_free_num --;
	LOG("object data startIdx %u,free startIdx %u,free page num %u,total page num %u\n",
		pObjBlk[puser->objIdx].data_startIdx,pclass->data_page_free_startIdx,
		pclass->data_page_free_num,pclass->data_page_total_num);
	oid = puser->objIdx;
    }
    else
    {// consumer
	LOG("user %p objIdx:%u,free startIdx:%u,first readIdx:%u\n",puser,puser->objIdx,
		pclass->object_free_startIdx,pclass->object_first_readIdx);
	if(pclass->flags&MOBS_FLAG_PRIVATE)
	{// skip to next for other consumers to deal only in private mode
	    puser->objIdx = pclass->object_first_readIdx;
	}

	if(pclass->object_free_startIdx != pclass->object_first_readIdx && 
		pclass->object_free_startIdx!=puser->objIdx)
	{// if empty object list
	    if(pObjBlk[puser->objIdx].state != OBJECT_FLAG_COMPLETED)
	    {// maybe is writing this object now
		LOG("objIdx %u,state: 0x%02x,data size:%u\n",
			puser->objIdx,pObjBlk[puser->objIdx].state,pObjBlk[puser->objIdx].data_size);
		oid = MOBS_ERR_EMPTY;
	    }
	    else
	    {
		if(pclass->flags&MOBS_FLAG_PRIVATE)
		{// skip to next for other consumers to deal only in private mode
		    // and when finish deal,we must delete this puser->objIdx in Finish API.
		    pclass->object_first_readIdx = (puser->objIdx+1) % pclass->object_max_num;
		}
		pObjBlk[puser->objIdx].refCnt++;
		puser->datOffset = 0;
		oid = puser->objIdx;
		LOG("objIdx %u,left object num now %d\n",puser->objIdx,get_available_object_num(pclass,puser->objIdx));
	    }
	}
	else
	{
	    //LOG("free startIdx:%u,first readIdx:%u\n",pclass->object_free_startIdx,pclass->object_first_readIdx);
	    oid = MOBS_ERR_EMPTY;
	}
    }
    _unlock(pclass);
    return oid;
}

int mobs_write(uintptr_t uid,int oid,void *data,size_t count)
{
    if(data==NULL || count == 0) return MOBS_ERR_PARAM;

    CONVERT_UID_TO_PTR(uid);CHECK_OBJECT_ID(oid);

    ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)(pHeadAddr + pclass->object_zone_offset);
    ObjectBlockAttr *pObj = &pObjBlk[oid];

    size_t left = count;
    uint8_t *pDataAddr = (uint8_t*)(pHeadAddr + pclass->data_zone_offset);
    uint8_t *pDstAddr = NULL;
    uint32_t idx = pObj->data_startIdx;
    while(left > 0)
    {
	pDstAddr = pDataAddr + idx * pclass->data_page_size;
	size_t len = (left > pclass->data_page_size)?pclass->data_page_size:left;
	LOG("--- dstAddr %p,page idx %u,len = %zu\n",pDstAddr,idx,len);
	memcpy(pDstAddr,data+(count-left),len);
	left -= len;
	pObj->data_size += len;
	pObj->data_idxNum++;
	pObj->data_endIdx = idx;

	if(left > 0) // allocate another page to store data
	{
	    _lock(pclass);
	    if(pclass->data_page_free_num == 0)
	    {
		delete_object(pclass,-1);
	    }
	    idx = pclass->data_page_allocate_table[idx] = pclass->data_page_free_startIdx;
	    pclass->data_page_free_startIdx = pclass->data_page_allocate_table[pclass->data_page_free_startIdx];
	    pclass->data_page_free_num --;
	    _unlock(pclass);
	}
    }
    return count;
}

int mobs_read( uintptr_t uid,int oid,void *data,size_t count)
{
    if(data == NULL || count == 0)
	return MOBS_ERR_PARAM;

    CONVERT_UID_TO_PTR(uid);CHECK_OBJECT_ID(oid);

    ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)(pHeadAddr + pclass->object_zone_offset);
    ObjectBlockAttr *pObj = &pObjBlk[oid];

    if(pObj->state != OBJECT_FLAG_COMPLETED)
    {// maybe covered because of loop-delete op
	LOG("object %d state err:0x%02x\n",oid,pObj->state);
	return MOBS_ERR_INTERRUPTED;
    }

    LOG("want count %zu,data size %u\n",count,pObj->data_size);

    size_t left = (count > pObj->data_size)?pObj->data_size:count;
    size_t offset = 0;
    uint8_t *pDataAddr = (uint8_t*)(pHeadAddr + pclass->data_zone_offset);
    uint8_t *pDstAddr = NULL,*pEndPage = NULL;
    uint32_t idx = pObj->data_startIdx;
    while(left > 0)
    {
	// puser->datOffset is Obejct Total Data offset
	pDstAddr = pDataAddr + idx * pclass->data_page_size + puser->datOffset % pclass->data_page_size; // we need add offset in page
	pEndPage = pDataAddr + idx * pclass->data_page_size + pclass->data_page_size;	
	size_t len = pEndPage - pDstAddr;
	LOG("### dstAddr %p,idx %u,page blank %zu,data left %zu\n",pDstAddr,idx,len,left);
	if(len > left) len = left;
	if(len > 0)
	{
	    memcpy(data+offset,pDstAddr,len);
	    LOG("copy len %zu,offset = %zu,[0] = %c\n",len,offset,(char)pDstAddr[0]);
	    puser->datOffset += len;
	    offset += len;
	    left -= len;
	}

	idx = pclass->data_page_allocate_table[idx];
    }
    return offset;
}

int mobs_finish(uintptr_t uid,int oid)
{
    CONVERT_UID_TO_PTR(uid);CHECK_OBJECT_ID(oid);

    ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)(pHeadAddr + pclass->object_zone_offset);
    ObjectBlockAttr *pObj = &pObjBlk[oid];

    _lock(pclass);
    if(puser->role == ROLE_PRODUCER)
    {
	pObj->state = OBJECT_FLAG_COMPLETED;
    }
    else
    {// consumer next object
	if(pclass->flags&MOBS_FLAG_PRIVATE)
	{ // when finish deal,we must delete this puser->objIdx in Finish API.
	    delete_object(pclass,puser->objIdx);
	}
	puser->objIdx = (puser->objIdx+1) % pclass->object_max_num;
	puser->datOffset = 0;
    }
    pObj->refCnt --;
    _unlock(pclass);

    LOG("user 0x%08lx fini obj %d op,state: 0x%02x, refcnt: %hhu,data size:%u,start idx:%u,end idx: %u,idxNum:%u\n",
	    uid,oid,pObj->state,pObj->refCnt,pObj->data_size,pObj->data_startIdx,pObj->data_endIdx,pObj->data_idxNum);
    return 0;
}

int mobs_cancel(uintptr_t uid,int oid)
{
    CONVERT_UID_TO_PTR(uid);CHECK_OBJECT_ID(oid);

    ObjectBlockAttr *pObjBlk = (ObjectBlockAttr*)(pHeadAddr + pclass->object_zone_offset);
    ObjectBlockAttr *pObj = &pObjBlk[oid];

    _lock(pclass);
    if(puser->role == ROLE_PRODUCER)
    {
	pObj->state = OBJECT_FLAG_INCOMPLETE;
    }
    else
    {// consumer next object
	if(pclass->flags&MOBS_FLAG_PRIVATE)
	{ // when finish deal,we must delete this puser->objIdx in Finish API.
	    delete_object(pclass,puser->objIdx);
	}
	puser->objIdx = (puser->objIdx+1) % pclass->object_max_num;
	puser->datOffset = 0;
    }
    pObj->refCnt --;
    _unlock(pclass);

    LOG("user 0x%08lx cancel obj %d op,state: 0x%02x, refcnt: %hhu,data size:%u,start idx:%u,end idx: %u,idxNum:%u\n",
	    uid,oid,pObj->state,pObj->refCnt,pObj->data_size,pObj->data_startIdx,pObj->data_endIdx,pObj->data_idxNum);
    return 0;
}

